#include "server.h"

namespace srpc
{
    int TcpRpcChannelServer::Start()
    {
        evutil_socket_t listener;
        if (listener = socket(AF_INET, SOCK_STREAM, 0), listener < 0)
        {
            std::cerr << "get socket failed" << std::endl;
            return 1;
        }

        evutil_make_listen_socket_reuseable(listener);

        struct sockaddr_in sin;
        sin.sin_family = AF_INET;
        sin.sin_addr.s_addr = 0;
        sin.sin_port = htons(mPort);

        if (bind(listener, (struct sockaddr *)&sin, sizeof(sin)) < 0)
        {
            std::cerr << "bind socket failed" << std::endl;
            return 1;
        }

        if (listen(listener, mQueueSize) < 0)
        {
            std::cerr << "listen socket failed" << std::endl;
            return 1;
        }

        evutil_make_socket_nonblocking(listener);

        struct event_base *base = event_base_new();

        if (base == nullptr)
        {
            std::cerr << "init event_base failed" << std::endl;
            return 1;
        }
        struct event *listen_event;
        mpArgs->pEventBase = base;
        mpArgs->pServicePool = mpServicePool;
        listen_event = event_new(base, listener, EV_READ | EV_PERSIST, &TcpRpcChannelServer::do_accept, (void *)mpArgs);
        event_add(listen_event, NULL);
        event_base_dispatch(base);
        return 0;
    }

    void TcpRpcChannelServer::do_accept(evutil_socket_t listener, short event, void *arg)
    {
        pArgs args = (pArgs)arg;
        struct event_base *base = args->pEventBase;
        evutil_socket_t fd;
        struct sockaddr_in sin;
        socklen_t slen;
        fd = accept(listener, (struct sockaddr *)&sin, &slen);
        if (fd < 0)
        {
            std::cerr << "accept failed" << std::endl;
            return;
        }
        if (fd > FD_SETSIZE)
        {
            std::cerr << "fd > FD_SETSIZE" << std::endl;
            return;
        }

        printf("ACCEPT: fd = %u\n", fd);
        struct bufferevent *bev = bufferevent_socket_new(base, fd, BEV_OPT_CLOSE_ON_FREE);
        bufferevent_setcb(bev, &TcpRpcChannelServer::read_cb, NULL, TcpRpcChannelServer::error_cb, arg);
        bufferevent_enable(bev, EV_READ | EV_WRITE | EV_PERSIST);
    }

    void TcpRpcChannelServer::error_cb(struct bufferevent *bev, short event, void *arg)
    {
        evutil_socket_t fd = bufferevent_getfd(bev);
        printf("fd = %u, ", fd);
        if (event & BEV_EVENT_TIMEOUT)
        {
            printf("Timed out\n"); //if bufferevent_set_timeouts() called
        }
        else if (event & BEV_EVENT_EOF)
        {
            printf("connection closed\n");
        }
        else if (event & BEV_EVENT_ERROR)
        {
            printf("some other error\n");
        }
        bufferevent_free(bev);
    }

    void TcpRpcChannelServer::read_cb(struct bufferevent *bev, void *arg)
    {
        pArgs args = (pArgs)arg;
        EV_BASE base = args->pEventBase;
        ServicePool *pServicePool = args->pServicePool;
        const int MAX_SIZE = 1024;
        char line[MAX_SIZE + 1];
        int n;
        evutil_socket_t fd = bufferevent_getfd(bev);
        // TODO 读数据有bug
        n = bufferevent_read(bev, line, MAX_SIZE);
        line[n] = '\0';
        srpc::RpcMessagePB rpcMessagePB;
        rpcMessagePB.ParseFromString(line);
        std::string serviceName = rpcMessagePB.service();

        printf("Recive message from socket fd = %u, service name: %s \n", fd, serviceName.c_str());

        ServicePoolIter iter = pServicePool->find(serviceName);
        if (iter == pServicePool->end())
        {
            char buffer[1024];
            snprintf(buffer, 1024, "service name %s is not registed", serviceName.c_str());
            std::cout << buffer << std::endl;
            return;
        }

        google::protobuf::Service *service = iter->second;
        const google::protobuf::MethodDescriptor *methodDesc = service->GetDescriptor()->FindMethodByName(rpcMessagePB.method());
        google::protobuf::Message *request = service->GetRequestPrototype(methodDesc).New();
        google::protobuf::Message *response = service->GetResponsePrototype(methodDesc).New();
        request->ParseFromString(rpcMessagePB.request());

        service->CallMethod(methodDesc, NULL, request, response, NULL);
        rpcMessagePB.clear_request();
        rpcMessagePB.set_type(srpc::RPC_TYPE_RESPONSE);
        rpcMessagePB.set_response(response->SerializeAsString());
        std::string rpcMessagePBStr = rpcMessagePB.SerializeAsString();
        evbuffer_drain(bev->output, evbuffer_get_length(bev->output)); // 清空缓存
        int status = bufferevent_write(bev, rpcMessagePBStr.c_str(), rpcMessagePBStr.size());
        if (status == 0)
        {
            std::cout << "send ok" << std::endl;
            std::cout << rpcMessagePB.DebugString()  << std::endl;
        }
        else
        {
            std::cout << "send failed" << std::endl;
        }
    }

    void TcpRpcChannelServer::write_cb(struct bufferevent *bev, void *arg)
    {
    }

    void TcpRpcChannelServer::RegisterService(google::protobuf::Service *service)
    {
        (*mpServicePool)[service->GetDescriptor()->name()] = service;
    }

    void TcpRpcChannelServer::Close()
    {
    }
}
